package hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.jdbc.PhoenixDriver;
//import org.apache.phoenix.jdbc.PhoenixDriver;
import java.io.IOException;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Iterator;

//HBase Java API使用IDEA开发
//https://blog.csdn.net/jinxing_000/article/details/130039092

public class ReadHbase {

    public static void main(String[] args) throws SQLException {
        //设置日志路径
        String logDir = System.getProperty("log.dir");

        if (logDir == null) {
            // 开发环境中使用
            logDir = System.getProperty("user.dir")+"/logs/";
            System.setProperty("log.dir", logDir);
        }
        System.out.println("系统日志路径: " + System.getProperty("log.dir"));

        Configuration conf=null;
        Connection conn=null;
        Admin admin=null;
        conf= HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");
        //conf.addResource("core-site.xml");
        //conf.addResource("hbase-site.xml");
        try {
            conn = ConnectionFactory.createConnection(conf);
            //读取表
            Table mytable = conn.getTable(TableName.valueOf("split_table_test"));
            //全表扫描
            //从第一个到最后一个 stop设置为空
            Scan scan=new Scan();
            //查询rowid是1开始的所有记录，注释下面两句就是全表搜索。
            scan.withStartRow("1009".getBytes());   //withStartRow不包含"1009"，即 >'1009'
            scan.withStopRow("1009|".getBytes());  //withStopRow包含"1009|"，即 <='1009|'
            ResultScanner scanner = mytable.getScanner(scan);
            //result  是与一行数据（有多个列族，多个列）
            Iterator<Result> iterator =  scanner.iterator();
            System.out.println("rowid"+"\t\t"+"id"+"\t\t"+"name");
            int cnt=0;
            while(iterator.hasNext()){
                Result result = iterator.next();
                //获取结果 返回的是cell 用cellutils将其变为字节数组 然后再转为string
                String rowid = Bytes.toString(result.getRow());
                String id = new String(CellUtil.cloneValue(
                        result.getColumnLatestCell("cf".getBytes(),"id".getBytes())));
                String name = new String(CellUtil.cloneValue(
                        result.getColumnLatestCell("cf".getBytes(),"name".getBytes())));
                System.out.println(rowid+"\t"+id+"\t"+name);
                cnt++;
            }
            System.out.println("普通方式的数据遍历成功！rowid为1009开始的记录数量="+cnt);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        java.sql.Connection connection=null;
        try {
            int cnt=0;
            //Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
            //等价：Class.forName(PhoenixDriver.class.getName());
            Class.forName(PhoenixDriver.class.getName());
            System.out.println("phoenix连接开始...");
            connection = DriverManager.getConnection("jdbc:phoenix:node1,node2,node3:2181");
            System.out.println("phoenix连接成功，数据遍历开始");
            //查询rowid 以1开始的所有数据
            String sql = String.format("select * from \"%s\" where \"rowid\" like '%s'","split_table_test","1009%");  //"select * from \"split_table_test\" where \"";

            PreparedStatement preparedStatement = connection.prepareStatement(sql);
            ResultSet resultSet = preparedStatement.executeQuery();

            System.out.println("row id"+"\t\t"+"id"+"\t"+"name");
            while (resultSet.next()) {
                cnt++;
                System.out.println(resultSet.getString("rowid")+"\t"+resultSet.getString("id")+"\t"+resultSet.getString("name"));
            }
            System.out.println("phoenix数据遍历成功！rowid为1009开始的记录数量="+cnt);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException("加载Phoenix驱动失败!");
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("获取Phoenix JDBC连接失败!");
        } finally {
            if (connection != null) {
                connection.close();
            }
        }
    }
}